Source code for hysop.backend.host.host_buffer

# Copyright (c) HySoP 2011-2024
#
# This file is part of HySoP software.
# See "https://particle_methods.gricad-pages.univ-grenoble-alpes.fr/hysop-doc/"
# for further info.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.


import numpy as np
import ctypes as C

from hysop.constants import MemoryOrdering, default_order
from hysop.tools.htypes import check_instance
from hysop.core.memory.buffer import Buffer, PooledBuffer




class HostBuffer(np.ndarray, Buffer):
    """
    Host buffer class.
    """

    __array_priority__ = -1.0

    def __new__(
        cls,
        size,
        shape=None,
        dtype=np.uint8,
        order=None,
        buffer=None,
        offset=0,
        strides=None,
    ):

        from_buffer = False
        if isinstance(buffer, Buffer):
            __buffer = buffer
            buffer = buffer.buf
            from_buffer = True

        obj = super().__new__(
            cls,
            shape=shape or (size,),
            dtype=dtype,
            order=order,
            buffer=buffer,
            offset=offset,
            strides=strides,
        )

        # keep a reference to the buffer (usefull for pooled buffers)
        # such that buffer.__del__ will only be called when all views
        # on this HostBuffer have been destroyed.
        if from_buffer and isinstance(__buffer, HostPooledBuffer):
            obj._hysop_base_data = __buffer

        return obj

    def __array_finalize__(self, obj):
        if (obj is not None) and isinstance(obj, HostBuffer):
            self._hysop_base_data = getattr(obj, "_hysop_base_data", None)

    def __str__(self):
        return self.view(np.ndarray).__str__()

    def __repr__(self):
        return self.view(np.ndarray).__repr__()



    def get_int_ptr(self):
        return self.ctypes.data




    def release(self):
        pass




    @classmethod
    def from_int_ptr(cls, int_ptr_value, size):
        """
        Given int ptr should never be freed, numpy take ownership.
        """
        buf = np.ctypeslib.as_array(
            C.cast(int_ptr_value, C.POINTER(C.c_uint8)), (size,)
        )
        return cls.from_buffer(buf)




    @classmethod
    def from_buffer(cls, buf):
        buf = np.frombuffer(buf, dtype=np.uint8)
        return cls._wrap(buf)




    def aligned_view(self, alignment, size=None):
        assert self.ndim == 1
        assert self.dtype == np.uint8
        assert alignment > 0
        assert not (alignment & (alignment - 1)), "alignment is not a power of 2."
        ptr = self.get_int_ptr()
        offset = -ptr % alignment
        if size is None:
            size = self.size - offset
        else:
            assert self.size >= (offset + size)
        buf = self.__getitem__(slice(offset, offset + size))
        return buf


    @classmethod
    def _wrap(cls, handle):
        assert isinstance(handle, np.ndarray)
        assert handle.ndim == 1
        assert handle.dtype == np.uint8
        return handle.view(cls)

    int_ptr = property(get_int_ptr)





class HostPooledBuffer(PooledBuffer):


    def get_array(self):
        return self._bufview


    array = property(get_array)